package com.quanyan.stadium.component.mq;

import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.quanyan.common.constants.MqConstants;
import com.quanyan.common.utils.ObjectUtil;
import com.quanyan.mq.enums.MqConsumerStatus;
import com.quanyan.mq.mq.consumer.AbstractMqConsumer;
import com.quanyan.mq.mq.consumer.MqTopic;
import com.quanyan.stadium.service.commonComponent.DataReductionService;
import com.quanyan.statium.mq.UserRegisterMsg;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;

/**
 * Created by xiaoxiao on 16/6/13.
 */
@Component(value = "stadiumUserRegisterConsumer")
public class StadiumUserRegisterConsumer extends AbstractMqConsumer {

    private final static Logger logger = LoggerFactory.getLogger(StadiumUserRegisterConsumer.class);

    @Value("${mq.nameServer.address}")
    private String nameServerAddress;

    @Value("${mq.nameServer.consumerUserGroup}")
    private String producerGroup;

    @Value("${mq.nameServer.consumerUserInstanceName}")
    private String instanceName;

    @Autowired
    private DataReductionService dataReductionService;

    @PostConstruct
    void run() {
        super.start();
    }

    @Override
    protected MqConsumerStatus process(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
        /**
         * 实现自己的逻辑业务
         *
         * */
        /**
         *  默认msgs里只有一条消息，可以通过设置consumeMessageBatchMaxSize参数来批量接收消息
         *  consumer.setConsumeMessageBatchMaxSize(X);
         */
        MessageExt msg = messages.get(0);
        String msgId = msg.getMsgId();
        byte[] bodyByte = msg.getBody();
        UserRegisterMsg userRegisterMsg = (UserRegisterMsg) ObjectUtil.toObject(bodyByte);
        logger.info("运动场消费用户注册消息，msgId：{}, 消息主题: {}, 消息tag: {}, 消息内容：{} ", msgId, msg.getTopic(), msg.getTags(), bodyByte);
        try{
            dataReductionService.initUserCategoryScore(userRegisterMsg.getUid(), userRegisterMsg.getCityId());
        }catch (Exception e) {
            logger.error("运动场消费用户注册消息,当前执行的topic:{},当前执行的tag:{}错误原因:{}",
                    msg.getTopic(), msg.getTags(), e);
        }
        return MqConsumerStatus.CONSUMER_SUCCESS;
    }

    @Override
    protected String getConsumerGroup() {
        return this.producerGroup;
    }

    @Override
    protected String getNameServerAddress() {
        return nameServerAddress;
    }

    @Override
    protected String getInstanceName() {
        return this.instanceName;
    }

    @Override
    protected MqTopic[] getMqTopics() {
        MqTopic[] mqTopics = new MqTopic[1];
        mqTopics[0] = new MqTopic(MqConstants.StadiumUserRegisterConsumer.STADIUM_USER_REGISTER_TOPIC, MqConstants.StadiumUserRegisterConsumer.STADIUM_USER_REGISTER_TAGS);
        return mqTopics;
    }
}
